/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.ban.data.exchange.runner.master;

import java.util.Set;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.sysflag.TopicSysFlag;
import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
import org.apache.rocketmq.tools.command.CommandUtil;
import org.apache.log4j.Logger;

public class CreateTopicTools {

    static Logger logger = Logger.getLogger(CreateTopicTools.class);

    public static void execute(String topicName, String cluster, String nameSrv) {
        //updateTopic -n 10.136.64.29:9877 -c Cluster1 -r 4 -w 4 -t twb_user_pic_mq
        DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt();
        defaultMQAdminExt.setNamesrvAddr(nameSrv);
        defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
        try {
            TopicConfig topicConfig = new TopicConfig();
            topicConfig.setReadQueueNums(8);
            topicConfig.setWriteQueueNums(8);
            topicConfig.setTopicName(topicName);

            boolean isUnit = false;
            boolean isCenterSync = false;
            int topicCenterSync = TopicSysFlag.buildSysFlag(isUnit, isCenterSync);
            topicConfig.setTopicSysFlag(topicCenterSync);
            boolean isOrder = false;
            topicConfig.setOrder(isOrder);

            defaultMQAdminExt.start();

            Set<String> masterSet = null;
            while (true) {
                try {
                    masterSet
                            = CommandUtil.fetchMasterAddrByClusterName(defaultMQAdminExt, cluster);
                    break;
                } catch (Exception e) {
                    logger.error("fetch master addr by clustername " + cluster + " failed, cause " + e.getMessage(), e);
                }
                Thread.sleep(100);
            }

            for (String addr : masterSet) {
                int num = 0;
                while (num < 3) {
                    try {
                        num++;
                        defaultMQAdminExt.createAndUpdateTopicConfig(addr, topicConfig);
                        logger.debug("create topic " + topicName + " success at " + addr);
                        break;
                    } catch (Exception e) {
                        if (num == 3) {
                            logger.error("create topic " + topicName + " at " + addr + " failed, cause " + e.getMessage(), e);
                        }
                        Thread.sleep(100);
                    }
                }
            }
        } catch (Exception e) {
            logger.error("create topic " + topicName + " failed, cause " + e.getMessage(), e);
        } finally {
            defaultMQAdminExt.shutdown();
        }
    }
}
